package com.yesep.learn.mqtt.client.handler;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.yesep.learn.mqtt.client.common.event.EventType;
import com.yesep.learn.mqtt.client.common.util.ChannelUtil;
import com.yesep.learn.mqtt.client.handler.protocol.ConnAck;
import com.yesep.learn.mqtt.client.handler.protocol.PingResq;
import com.yesep.learn.mqtt.client.handler.protocol.PubAck;
import com.yesep.learn.mqtt.client.handler.protocol.PubComp;
import com.yesep.learn.mqtt.client.handler.protocol.PubRec;
import com.yesep.learn.mqtt.client.handler.protocol.PubRel;
import com.yesep.learn.mqtt.client.handler.protocol.Publish;
import com.yesep.learn.mqtt.client.handler.protocol.SubAck;
import com.yesep.learn.mqtt.client.handler.protocol.UnSubAck;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.codec.UnsupportedMessageTypeException;
import io.netty.handler.codec.mqtt.MqttConnAckMessage;
import io.netty.handler.codec.mqtt.MqttMessage;
import io.netty.handler.codec.mqtt.MqttPubAckMessage;
import io.netty.handler.codec.mqtt.MqttPublishMessage;
import io.netty.handler.codec.mqtt.MqttSubAckMessage;
import io.netty.handler.codec.mqtt.MqttUnsubAckMessage;

/**
 * 
 * @Description:
 */
public class MqttClientHandler extends ChannelInboundHandlerAdapter {

	private Logger logger = LoggerFactory.getLogger(getClass());
	private ConnAck connAck;
	private SubAck subAck;
	private UnSubAck unSubAck;
	private Publish publish;
	private PingResq pingResq;
	private PubRel pubRel;
	private PubAck pubAck;
	private PubRec pubRec;
	private PubComp pubComp;

	public MqttClientHandler() {
		connAck = new ConnAck();
		subAck = new SubAck();
		unSubAck = new UnSubAck();
		publish = new Publish();
		pingResq = new PingResq();
		pubRel = new PubRel();
		pubAck = new PubAck();
		pubRec = new PubRec();
		pubComp = new PubComp();
	}

	public MqttClientHandler(ConnAck connAck, SubAck subAck, UnSubAck unSubAck, Publish publish, PubRel pubRel, PubAck pubAck, PubRec pubRec, PubComp pubComp, PingResq pingResq) {
		this.connAck = connAck;
		this.subAck = subAck;
		this.unSubAck = unSubAck;
		this.publish = publish;
		this.pingResq = pingResq;
		this.pubRel = pubRel;
		this.pubAck = pubAck;
		this.pubRec = pubRec;
		this.pubComp = pubComp;
	}

	@Override
	public void channelRead(ChannelHandlerContext ctx, Object message) throws Exception {
		MqttMessage msg = (MqttMessage) message;
		switch (msg.fixedHeader().messageType()) {
		case CONNECT://请求连接
			break;
		case CONNACK://请求应答
			connAck.process(ctx, (MqttConnAckMessage) msg);
			break;
		case PUBLISH://发布消息
			publish.process(ctx, (MqttPublishMessage) msg);
			break;
		case PUBACK://发布应答
			pubAck.process(ctx, (MqttPubAckMessage) msg);
			break;
		case PUBREC://发布已接收回执
			pubRec.process(ctx, msg);
			break;
		case PUBREL://发布释放
			pubRel.process(ctx, msg);
			break;
		case PUBCOMP://发布完成
			pubComp.process(ctx, msg);
			break;
		case SUBSCRIBE://订阅请求
			break;
		case SUBACK://订阅应答
			subAck.process(ctx, (MqttSubAckMessage) msg);
			break;
		case UNSUBSCRIBE://取消订阅
			break;
		case UNSUBACK://取消订阅应答
			unSubAck.process(ctx, (MqttUnsubAckMessage) msg);
			break;
		case PINGREQ://ping请求
			break;
		case PINGRESP://ping请求回应
			pingResq.process(ctx.channel(), msg);
			break;
		case DISCONNECT://断开连接
			break;
		default:
			throw new UnsupportedMessageTypeException("Unacceptable Message Type");
		}
	}

	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
		ctx.close();
		logger.error("bad thing happened : ", cause);
	}

	@Override
	public void channelInactive(ChannelHandlerContext ctx) throws Exception {
		ChannelUtil.triggerUserEvent(ctx, EventType.DIS_CONNECT, null);
		logger.info("lose connection : {}", ChannelUtil.clientId(ctx.channel()));
	}

}
